package com.atguigu;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.yarn.webapp.hamlet.Hamlet;

import java.io.IOException;

public class Hbase_DML {
    //获取到Hbase连接
    public static Connection connection = HbaseConnectionSingle.connection;

    //插入数据
    public static void putCell(String namespace,String tableName,String rowKey,String family,String column,String value) throws IOException {
        //1.通过Connection获取到Table实例
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        //2.创建Put对象
        Put put = new Put(Bytes.toBytes(rowKey));

        //2.1指定要往那一个列族的哪一列中插入什么样的数据
//        put.addColumn(family.getBytes(), column.getBytes(), value.getBytes());
        put.addColumn(Bytes.toBytes(family),Bytes.toBytes(column),Bytes.toBytes(value));

        //3.调用put方法写入数据
        table.put(put);

        //4.关闭Table
        table.close();
    }

    //获取数据
    public static void getCells(String namespace,String tableName,String rowKey,String family,String column) throws IOException {
        //1.通过Connection获取Table实例
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        //2.创建Get对象
        Get get = new Get(Bytes.toBytes(rowKey));

        //2.1指定获取哪一个列族的数据
//        get.addFamily(Bytes.toBytes(family));

        //2.2指定获取哪一个列族哪一列的数据
//        get.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));

        //2.3获取指定版本个数的数据
//        get.readVersions(10);
        //2.4获取所有可用(维护的)版本的数据
        get.readAllVersions();

        //3.调用get方法读取数据
        Result result = table.get(get);

//        byte[] value = result.value();
//        System.out.println(Bytes.toString(value));

        //4获取到cell数组并从中获取到rowkey 列族 列 值 并打印到控制台
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            System.out.println(Bytes.toString(CellUtil.cloneRow(cell))+":"+Bytes.toString(CellUtil.cloneFamily(cell))+":"+Bytes.toString(CellUtil.cloneQualifier(cell))+":"+Bytes.toString(CellUtil.cloneValue(cell)));
        }

        //关闭Table
        table.close();

    }
    //删除数据
    public static void deleteCells(String namespace,String tableName,String rowKey,String family,String column) throws IOException {
        //1.通过Connection获取Table实例
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));

        //2.创建Delete对象，指定rowkey，可以删除一行数据
        Delete delete = new Delete(Bytes.toBytes(rowKey));

        //2.1指定删除某一列族的数据
        delete.addFamily(Bytes.toBytes(family));

        //2.2指定删除某一列的数据
//        delete.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));
        //2.3删除指定某一列所有版本的数据
//        delete.addColumns(Bytes.toBytes(family), Bytes.toBytes(column));


        //3.调用delete方法删除数据
        table.delete(delete);

        //4.关闭Table
        table.close();

    }
    //扫描数据
    public static void scanRows(String namespace,String tableName,String startRow,String stopRow) throws IOException {
        //1.通过Connection获取Table实例
        Table table = connection.getTable(TableName.valueOf(namespace, tableName));


        Scan scan = new Scan();

        //获取所有版本的数据（超过维护的版本）  默认读取列族所维护的版本数据
//        scan.setRaw(true);
        scan.readAllVersions();
        //指定读几个版本的数据
        scan.readVersions(3);


        //指定开始行和结束行，第二个boolean类型的参数代表的是是否包含当前行，true包含 false反之
        scan.withStartRow(Bytes.toBytes(startRow),true);
        scan.withStopRow(Bytes.toBytes(stopRow),true);



        ResultScanner scanner = table.getScanner(scan);

        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println(Bytes.toString(CellUtil.cloneRow(cell))+":"+Bytes.toString(CellUtil.cloneFamily(cell))+":"+Bytes.toString(CellUtil.cloneQualifier(cell))+":"+Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }


        //关闭Table
        table.close();

    }
    public static void main(String[] args) throws IOException {
//        putCell("bigdata", "student", "1007", "info", "age", "80");
//        putCell("bigdata", "student", "1007", "info", "name", "xiaoqi");
        putCell("bigdata", "student", "1002", "info", "age", "80");
        putCell("bigdata", "student", "1002", "info", "age", "81");
        putCell("bigdata", "student", "1002", "info", "age", "82");
        putCell("bigdata", "student", "1002", "info", "age", "83");
        putCell("bigdata", "student", "1002", "info", "age", "84");
        putCell("bigdata", "student", "1002", "info", "age", "85");
        putCell("bigdata", "student", "1002", "info", "age", "86");

//        deleteCells("bigdata", "student", "1002", "info", "age");

//        getCells("bigdata", "student", "1002", "info", "age");

        scanRows("bigdata", "student", "1002", "1006");
        //关闭Hbase链接
        HbaseConnectionSingle.closeConnection();
    }
}
